package com.yifeng.repo.base.mdm.full.worker;

import com.talkyun.utils.Looper;
import com.yifeng.base.bds.api.store.dto.StoreCoreInfoDto;
import com.yifeng.base.bds.api.store.dto.StoreCorporationRegionInfoDto;
import com.yifeng.base.bds.api.store.params.StoreInfoParam;
import com.yifeng.repo.base.constant.BaseConstant;
import com.yifeng.repo.base.mdm.MasterDataManager;
import com.yifeng.repo.base.mdm.full.holder.StoreRouteHolder;
import com.yifeng.repo.base.mdm.full.listener.FullDataListener;
import com.yifeng.repo.base.mdm.full.listener.impl.DefaultFullDataListener;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.List;

/**
 * Created by daibing on 2024/6/20.
 */
@Slf4j
public class FullDataWorker {
    private final MasterDataManager masterDataManager;
    private final FullDataListener fullDataListener;
    private Looper worker;

    public FullDataWorker(MasterDataManager masterDataManager) {
        this.masterDataManager = masterDataManager;
        this.fullDataListener = new DefaultFullDataListener();
    }

    public FullDataWorker(MasterDataManager masterDataManager, FullDataListener fullDataListener) {
        this.masterDataManager = masterDataManager;
        this.fullDataListener = fullDataListener != null ? fullDataListener : new DefaultFullDataListener();
    }

    public void init() {
        worker = buildWorker();
        worker.start();
    }

    public void destroy() {
        worker.shutdown();
    }

    /**
     * 3分钟运行一·次, 第一次拉取全量数据，后续仅拉取增量数据：
     * 1、每次拉取一个波次的数据，都是一个偏移时间，根据门店编号去遍历
     * 2、开始下一个波次，门店编号要复位，偏移时间要用上一次开始前的时间
     */
    private Looper buildWorker() {
        // 准备参数
        StoreInfoParam param = new StoreInfoParam();
        param.setOffsetTime(null);
        param.setOffsetStoreCode(null);
        param.setPageSize(BaseConstant.BASE_BATCH_SIZE_LIMIT);

        // 构建Looper
        return new Looper("full-data-worker", BaseConstant.THREE_MINUTE_MS.intValue(), BaseConstant.ONE_SECOND_MS.intValue()) {
            @Override
            protected void loop() throws Throwable {
                int count = 0;
                LocalDateTime startWaveTime = LocalDateTime.now();
                do {
                    // 1. 拉取数据
                    List<StoreCorporationRegionInfoDto> storeDtos =
                            masterDataManager.getStoreMasterDataRestService().listCorporationRegionInfo(param);
                    if (storeDtos.isEmpty()) {
                        break;
                    }

                    // 2. 更新offset和统计信息
                    count += storeDtos.size();
                    StoreCoreInfoDto last = storeDtos.get(storeDtos.size() - 1);
                    param.setOffsetStoreCode(last.getStoreCode());

                    // 3. 调整数据并且写入Holder
                    int rawSize = storeDtos.size();
                    StoreRouteHolder.add(fullDataListener.onStoreData(storeDtos));
                    if (rawSize < BaseConstant.BASE_BATCH_SIZE_LIMIT) {
                        break;
                    }
                } while (true);
                log.info("FullDataWorker success：offsetTime={}, startWaveTime={}, offsetStoreCode={}, count={}",
                        param.getOffsetTime(), startWaveTime, param.getOffsetStoreCode(), count);
                param.setOffsetTime(startWaveTime);
                param.setOffsetStoreCode(null);
            }

            @Override
            protected void loopThrowable(Throwable t) {
                log.warn("FullDataWorker failed: offsetTime={}, offsetStoreCode={}, error=",
                        param.getOffsetTime(), param.getOffsetStoreCode(), t);
            }
        };
    }

}
